package com.pengheng.transformation.aggregation;

import com.pengheng.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 按键分区
 */
public class TransKeyBy {
    public static void main(String[] args) throws Exception {
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 2L, 2),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
        );
        //通过 keyBy 将相同的数据 分配到同一个分区
        stream.keyBy(e -> e.id).print();
        /**
         * 2> WaterSensor{id='sensor_2', ts=2, vc=2}
         * 5> WaterSensor{id='sensor_1', ts=1, vc=1}
         * 5> WaterSensor{id='sensor_1', ts=2, vc=2}
         * 1> WaterSensor{id='sensor_3', ts=3, vc=3}
         */

        env.execute();




    }
}
